/**
 * Created by lzfto
 * Created time: 2025/05/25
 */
package org.lzfto.flink.demo.table.connector;

import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.COMPATIBLE_MODE;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.DRIVER;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.FILTER_HANDLING_POLICY;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.LOOKUP_CACHE_MAX_ROWS;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.LOOKUP_CACHE_MISSING_KEY;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.LOOKUP_CACHE_TTL;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.LOOKUP_MAX_RETRIES;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.MAX_RETRY_TIMEOUT;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.SCAN_AUTO_COMMIT;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.SCAN_FETCH_SIZE;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.SCAN_PARTITION_COLUMN;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.SCAN_PARTITION_LOWER_BOUND;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.SCAN_PARTITION_NUM;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.SCAN_PARTITION_UPPER_BOUND;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.SINK_MAX_RETRIES;
import static org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions.SINK_PARALLELISM;
import static org.apache.flink.connector.jdbc.utils.JdbcUtils.getConnectionProperties;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.jdbc.core.database.JdbcFactoryLoader;
import org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialect;
import org.apache.flink.connector.jdbc.core.table.source.JdbcDynamicTableSource;
import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions;
import org.apache.flink.connector.jdbc.internal.options.JdbcReadOptions;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.lookup.LookupOptions;
import org.apache.flink.table.connector.source.lookup.cache.DefaultLookupCache;
import org.apache.flink.table.connector.source.lookup.cache.LookupCache;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.Preconditions;
import org.lzfto.flink.demo.common.JdbcOptionRecord;
import org.lzfto.flink.demo.table.connector.sink.CustomJdbcSink;

/**
 * 自定义的jdbc连接器，仅仅是为了能够删除。
 * <br/>如果是为了在插入前删除，而没有其它操作，不建议这样自定义，因为这种情况下删除的操作实在低效得很！
 */
public class CustomJdbcConnectorFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {

	private static final ConfigOption<String> DRIVER_CLASS = ConfigOptions.key("driver-class").stringType().noDefaultValue();
	private static final ConfigOption<String> URL = ConfigOptions.key("url").stringType().noDefaultValue();
	private static final ConfigOption<String> TABLE_NAME = ConfigOptions.key("table-name").stringType()
			.noDefaultValue();
	private static final ConfigOption<String> USERNAME = ConfigOptions.key("username").stringType().noDefaultValue();
	private static final ConfigOption<String> PASSWORD = ConfigOptions.key("password").stringType().noDefaultValue();
	//
	private static final ConfigOption<String> IDCOLCODE = ConfigOptions.key("idColCode").stringType().noDefaultValue();
	@Override
	public DynamicTableSink createDynamicTableSink(Context context) {
		return this.createSink(context);
	}

	private DynamicTableSink createSink(Context context) {
		DataType dataType=context.getPhysicalRowDataType();
		final ReadableConfig options = FactoryUtil.createTableFactoryHelper(this, context).getOptions();
		JdbcOptionRecord jdbcOptionRecord=new JdbcOptionRecord(
				options.get(DRIVER_CLASS),
				options.get(URL), 
				options.get(TABLE_NAME), 
				options.get(USERNAME),
				options.get(PASSWORD),
				options.get(IDCOLCODE),
				dataType);
		return new CustomJdbcSink(jdbcOptionRecord);
	}



	@Override
	public String factoryIdentifier() {
		return "custom-jdbc";
	}

	@Override
	public Set<ConfigOption<?>> requiredOptions() {
		Set<ConfigOption<?>> options = new HashSet<>();
		options.add(URL);
		options.add(TABLE_NAME);
		options.add(USERNAME);
		options.add(PASSWORD);
		options.add(IDCOLCODE);
		options.add(DRIVER_CLASS);
		return options;
	}

	@Override
	public Set<ConfigOption<?>> optionalOptions() {
		return Collections.emptySet();
	}

	@Override
	public DynamicTableSource createDynamicTableSource(Context context) {
		final FactoryUtil.TableFactoryHelper helper =
				FactoryUtil.createTableFactoryHelper(this, context);
		final ReadableConfig config = helper.getOptions();

		helper.validate();
		validateConfigOptions(config, context.getClassLoader());
		validateDataTypeWithJdbcDialect(
				context.getPhysicalRowDataType(),
				config.get(URL),
				config.get(COMPATIBLE_MODE),
				context.getClassLoader());
		return new JdbcDynamicTableSource(
				getJdbcOptions(helper.getOptions(), context.getClassLoader()),
				getJdbcReadOptions(helper.getOptions()),
				helper.getOptions().get(LookupOptions.MAX_RETRIES),
				getLookupCache(config),
				helper.getOptions().get(FILTER_HANDLING_POLICY),
				context.getPhysicalRowDataType());
	}

	@Override
	public Set<ConfigOption<?>> forwardOptions() {
		return DynamicTableSinkFactory.super.forwardOptions();
	}

	@SuppressWarnings("deprecation")
	private void validateConfigOptions(ReadableConfig config, ClassLoader classLoader) {
		String jdbcUrl = config.get(URL);
		JdbcFactoryLoader.loadDialect(jdbcUrl, classLoader, config.get(COMPATIBLE_MODE));

		checkAllOrNone(config, new ConfigOption[] {USERNAME, PASSWORD});

		checkAllOrNone(
				config,
				new ConfigOption[] {
						SCAN_PARTITION_COLUMN,
						SCAN_PARTITION_NUM,
						SCAN_PARTITION_LOWER_BOUND,
						SCAN_PARTITION_UPPER_BOUND
				});

		if (config.getOptional(SCAN_PARTITION_LOWER_BOUND).isPresent()
				&& config.getOptional(SCAN_PARTITION_UPPER_BOUND).isPresent()) {
			long lowerBound = config.get(SCAN_PARTITION_LOWER_BOUND);
			long upperBound = config.get(SCAN_PARTITION_UPPER_BOUND);
			if (lowerBound > upperBound) {
				throw new IllegalArgumentException(
						String.format(
								"'%s'='%s' must not be larger than '%s'='%s'.",
								SCAN_PARTITION_LOWER_BOUND.key(),
								lowerBound,
								SCAN_PARTITION_UPPER_BOUND.key(),
								upperBound));
			}
		}

		checkAllOrNone(config, new ConfigOption[] {LOOKUP_CACHE_MAX_ROWS, LOOKUP_CACHE_TTL});

		if (config.get(LOOKUP_MAX_RETRIES) < 0) {
			throw new IllegalArgumentException(
					String.format(
							"The value of '%s' option shouldn't be negative, but is %s.",
							LOOKUP_MAX_RETRIES.key(), config.get(LOOKUP_MAX_RETRIES)));
		}

		if (config.get(SINK_MAX_RETRIES) < 0) {
			throw new IllegalArgumentException(
					String.format(
							"The value of '%s' option shouldn't be negative, but is %s.",
							SINK_MAX_RETRIES.key(), config.get(SINK_MAX_RETRIES)));
		}

		if (config.get(MAX_RETRY_TIMEOUT).getSeconds() <= 0) {
			throw new IllegalArgumentException(
					String.format(
							"The value of '%s' option must be in second granularity and shouldn't be smaller than 1 second, but is %s.",
							MAX_RETRY_TIMEOUT.key(),
							config.get(
									ConfigOptions.key(MAX_RETRY_TIMEOUT.key())
											.stringType()
											.noDefaultValue())));
		}
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	private void checkAllOrNone(ReadableConfig config, ConfigOption<?>[] configOptions) {
		int presentCount = 0;
		for (ConfigOption configOption : configOptions) {
			if (config.getOptional(configOption).isPresent()) {
				presentCount++;
			}
		}
		String[] propertyNames =
				Arrays.stream(configOptions).map(ConfigOption::key).toArray(String[]::new);
		Preconditions.checkArgument(
				configOptions.length == presentCount || presentCount == 0,
				"Either all or none of the following options should be provided:\n"
						+ String.join("\n", propertyNames));
	}

	private static void validateDataTypeWithJdbcDialect(
			DataType dataType, String url, String compatibleMode, ClassLoader classLoader) {
		final JdbcDialect dialect = JdbcFactoryLoader.loadDialect(url, classLoader, compatibleMode);
		dialect.validate((RowType) dataType.getLogicalType());
	}

	private InternalJdbcConnectionOptions getJdbcOptions(
			ReadableConfig readableConfig, ClassLoader classLoader) {
		final String url = readableConfig.get(URL);
		final InternalJdbcConnectionOptions.Builder builder =
				InternalJdbcConnectionOptions.builder()
						.setClassLoader(classLoader)
						.setDBUrl(url)
						.setTableName(readableConfig.get(TABLE_NAME))
						.setDialect(
								JdbcFactoryLoader.loadDialect(
										url, classLoader, readableConfig.get(COMPATIBLE_MODE)))
						.setParallelism(readableConfig.getOptional(SINK_PARALLELISM).orElse(null))
						.setConnectionCheckTimeoutSeconds(
								(int) readableConfig.get(MAX_RETRY_TIMEOUT).getSeconds());

		readableConfig.getOptional(DRIVER).ifPresent(builder::setDriverName);
		readableConfig.getOptional(USERNAME).ifPresent(builder::setUsername);
		readableConfig.getOptional(PASSWORD).ifPresent(builder::setPassword);
		readableConfig.getOptional(COMPATIBLE_MODE).ifPresent(builder::setCompatibleMode);
		getConnectionProperties(readableConfig)
				.forEach((key, value) -> builder.setProperty(key.toString(), value.toString()));
		return builder.build();
	}

	@SuppressWarnings("deprecation")
	private LookupCache getLookupCache(ReadableConfig tableOptions) {
		LookupCache cache = null;
		// Legacy cache options
		if (tableOptions.get(LOOKUP_CACHE_MAX_ROWS) > 0
				&& tableOptions.get(LOOKUP_CACHE_TTL).compareTo(Duration.ZERO) > 0) {
			cache =
					DefaultLookupCache.newBuilder()
							.maximumSize(tableOptions.get(LOOKUP_CACHE_MAX_ROWS))
							.expireAfterWrite(tableOptions.get(LOOKUP_CACHE_TTL))
							.cacheMissingKey(tableOptions.get(LOOKUP_CACHE_MISSING_KEY))
							.build();
		}
		if (tableOptions
				.get(LookupOptions.CACHE_TYPE)
				.equals(LookupOptions.LookupCacheType.PARTIAL)) {
			cache = DefaultLookupCache.fromConfig(tableOptions);
		}
		return cache;
	}

	private JdbcReadOptions getJdbcReadOptions(ReadableConfig readableConfig) {
		final Optional<String> partitionColumnName =
				readableConfig.getOptional(SCAN_PARTITION_COLUMN);
		final JdbcReadOptions.Builder builder = JdbcReadOptions.builder();
		if (partitionColumnName.isPresent()) {
			builder.setPartitionColumnName(partitionColumnName.get());
			builder.setPartitionLowerBound(readableConfig.get(SCAN_PARTITION_LOWER_BOUND));
			builder.setPartitionUpperBound(readableConfig.get(SCAN_PARTITION_UPPER_BOUND));
			builder.setNumPartitions(readableConfig.get(SCAN_PARTITION_NUM));
		}
		readableConfig.getOptional(SCAN_FETCH_SIZE).ifPresent(builder::setFetchSize);
		builder.setAutoCommit(readableConfig.get(SCAN_AUTO_COMMIT));
		return builder.build();
	}
}
